package com.baixiaowen.xiaoaointerview.大厂笔试_白板面试篇.中文乱码处理;

import java.io.File;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;

/**
 * 大文件词频统计，多线程
 *
 *  阿姆达定律
 *  120s -> 16core -> 120/16 = ?
 *
 * @author Baixiaowen
 */
public class WordCount {

    final ForkJoinPool pool = ForkJoinPool.commonPool();

    private HashMap<String, Integer> countByString(String str) {
        HashMap<String, Integer> map = new HashMap<>();
        StringTokenizer tokenizer = new StringTokenizer(str);
        while (tokenizer.hasMoreTokens()) {
            String word = tokenizer.nextToken();
            incKey(word, map, 1);
        }
        return map;
    }

    private void incKey(String key, Map<String, Integer> map, int n) {
        if (map.containsKey(key)) {
            map.put(key, map.get(key) + 1);
        } else {
            map.put(key, n);
        }
    }

    /**
     * 并发任务
     */
    class CountTask implements Callable<HashMap<String, Integer>> {
        private final long start;
        private final long end;
        private final String fileName;

        public CountTask(String fileName, long start, long end) {
            this.start = start;
            this.end = end;
            this.fileName = fileName;
        }

        @Override
        public HashMap<String, Integer> call() throws Exception {
            HashMap<String, Integer> map = new HashMap<>();
            FileChannel channel = new RandomAccessFile(this.fileName, "rw").getChannel();

            // [start, end] -> Memory
            // Device -> Kernel Space -> User Space(Buffer) -> Thread
            // 将内核态内存临时提供给用户态使用 这里的 mbuf 就相当于一段内存区域
            MappedByteBuffer mbuf = channel.map(
                    FileChannel.MapMode.READ_ONLY,
                    this.start,
                    this.end - this.start
            );

            String str = StandardCharsets.US_ASCII.decode(mbuf).toString();
            return countByString(str);
        }
    }

    public void run(String fileName, long chunkSize) throws ExecutionException, InterruptedException {
        File file = new File(fileName);
        long fileSize = file.length();

        long position = 0;

        long startTime = System.currentTimeMillis();

        List<ForkJoinTask<HashMap<String, Integer>>> tasks = new ArrayList<>();
        while (position < fileSize) {
            long next = Math.min(position + chunkSize, fileSize);
            CountTask task = new CountTask(fileName, position, next);
            position = next;
            ForkJoinTask<HashMap<String, Integer>> future = pool.submit(task);
            tasks.add(future);
        }

        System.err.format("split to %d tasks\n", tasks.size());

        HashMap<String, Integer> totalMap = new HashMap<>();
        for (ForkJoinTask<HashMap<String, Integer>> future : tasks) {
            HashMap<String, Integer> map = future.get();
            for (Map.Entry<String, Integer> entry : map.entrySet()) {
                incKey(entry.getKey(), totalMap, entry.getValue());
            }
        }

        System.err.println("time: " + (System.currentTimeMillis() - startTime));

        System.err.println("total: " + totalMap.size());

        System.err.println(totalMap.get("ab"));
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        WordCount wordCount = new WordCount();
        wordCount.run("word.txt", 1024 * 1024);
    }

}
